package com.katesoft.scale4j.rttp.jobs.jdbc.init;

import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.ILock;
import net.jcip.annotations.ThreadSafe;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.perf4j.StopWatch;
import org.springframework.core.io.Resource;
import org.springframework.jdbc.datasource.init.ScriptStatementFailedException;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.Collection;
import java.util.LinkedHashSet;

import static com.katesoft.scale4j.rttp.internal.IRttpHazelcastRefs.SCHEDULER_QUARTZ_DB_UPDATE_LOCK;

/**
 * This class will create quartz schema during startup if necessary.
 * <p/>
 * This class is designed to be used within jvmcluster. first will obtain distributed lock and then create schema if necessary.
 *
 * @author kate2007
 */
@ThreadSafe
public class RttpQuartzSchemaCreator extends AbstractSchemaCreator
{
    public static final String SCHEDULER_DB_UPDATE_DONE = "rttp.scheduler.db.update.done";

    public RttpQuartzSchemaCreator()
    {
        tablesPrefix = "QRTZ_";
    }

    /**
     * will create scheduler schema if does not exists
     *
     * @param connection database connection
     *
     * @throws SQLException re-throw any SQLExceptions
     */
    @Override
    public void populate(Connection connection) throws SQLException
    {
        if (getSchedulerCreationScript() == null) { return; }
        StopWatch stopWatch = new StopWatch("rttp.quartz.db.create");
        HazelcastInstance instance = springHazelcastBridge.getHazelcastInstance();
        if (instance.getLifecycleService().isRunning()) {
            ILock lock = instance.getLock(SCHEDULER_QUARTZ_DB_UPDATE_LOCK);
            logger.info("using %s and dialect %s for scheduler schema creation", lock.toString(), dialect.toString());
            try {
                if (lock.tryLock() && !tablesCreated(connection)) {
                    if (!springHazelcastBridge.getAtomicBoolean(SCHEDULER_DB_UPDATE_DONE).get()) {
                        logger.info("scheduler schema is not created, will create using connection %s", connection);
                        springHazelcastBridge.updateAtomicBoolean(SCHEDULER_DB_UPDATE_DONE, true);
                        super.setScripts(new Resource[]{getSchedulerCreationScript()});
                        try {
                            super.populate(connection);
                        }
                        catch (ScriptStatementFailedException e) {
                            logger.warn("unable to create quartz schema for dialect %s, exception = %s", dialect.getSimpleName(),
                                        ExceptionUtils.getFullStackTrace(e));
                        }
                    }
                }
            }
            finally {
                lock.unlock();
                logger.debug("spring batch schema created in %s milisecs using %s", stopWatch.getElapsedTime(), connection);
            }
        }
    }

    @Override
    public Collection<String> tables()
    {
        Collection<String> tables = new LinkedHashSet<String>();
        tables.add(tablesPrefix + "JOB_DETAILS");
        tables.add(tablesPrefix + "JOB_LISTENERS");
        tables.add(tablesPrefix + "TRIGGERS");
        tables.add(tablesPrefix + "SIMPLE_TRIGGERS");
        tables.add(tablesPrefix + "CRON_TRIGGERS");
        tables.add(tablesPrefix + "BLOB_TRIGGERS");
        tables.add(tablesPrefix + "TRIGGER_LISTENERS");
        tables.add(tablesPrefix + "CALENDARS");
        tables.add(tablesPrefix + "PAUSED_TRIGGER_GRPS");
        tables.add(tablesPrefix + "FIRED_TRIGGERS");
        tables.add(tablesPrefix + "SCHEDULER_STATE");
        tables.add(tablesPrefix + "LOCKS");
        return tables;
    }
}
